package net.bwie.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.*;

/**
 * Flink DataStream采用同步方式，对流中数据进行补全，访问MySQL维表数据
 *      todo 使用DataStream中map算子即可
 * @author xuanyu
 * @date 2025/10/29
 */
public class _03OrderDimSync {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
/*
模拟数据：
u101,o1,milk,5.0,2025-10-29 17:30:10
u102,o2,water,2.0,2025-10-29 17:31:23
u103,o3,bread,6.0,2025-10-29 17:32:45
u104,o4,smoke,17.0,2025-10-29 17:33:17
 */
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3. 数据转换-transformation
		SingleOutputStreamOperator<String> stream1 = stream.map(
			new MapFunction<String, String>() {
				@Override
				public String map(String value) throws Exception {
					// 字符串分割 -> u101,o1,milk,5.0,2025-10-29 17:30:10
					String[] split = value.split(",");
					// 获取商品名称
					String skuName = split[2];
					// todo 请求数据库，获取商品sku对应品类category
					String categoryName = requestMysql(skuName);
					// 返回
					return value + "," + categoryName;
				}
			}
		);
		stream1.print("dwd");


		// 5. 触发执行-execute
		env.execute("OrderDimSync") ;
	}

	/**
	 * 请求MySQL，依据skuName获取categoryName名称
	 *      SELECT category_name FROM flink_day13.sku_info WHERE sku_name = ? ;
	 */
	private static String requestMysql(String skuName) throws Exception {
		// s0.变量
		Connection conn = null;
		PreparedStatement pstmt = null ;
		ResultSet resultSet = null;
		String categoryName = "unknown" ;
		try {
			// s1.连接
			conn = DriverManager.getConnection(
				"jdbc:mysql://node101:3306/flink_day13", "root", "123456"
			);
			// s2.语句
			pstmt = conn.prepareStatement(
				"SELECT category_name FROM flink_day13.sku_info WHERE sku_name = ?"
			);
			pstmt.setString(1, skuName);
			// s3-执行
			resultSet = pstmt.executeQuery();
			// s4-获取
			while (resultSet.next()){
				categoryName = resultSet.getString("category_name");
			}
		}catch (Exception e){
			e.printStackTrace();
		}finally {
			// s5.关闭
			resultSet.close();
			pstmt.close();
			conn.close();
		}
		return categoryName;
	}

}
